package com.audaque.springboot.foshanupload.web.rocketmqdemo.controller;

import com.audaque.springboot.foshanupload.core.result.R;
import com.audaque.springboot.foshanupload.web.rocketmqdemo.model.OrderPaidEvent;
import com.audaque.springboot.foshanupload.web.rocketmqdemo.model.User;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQLocalRequestCallback;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Slf4j
@RestController
@RequestMapping("/rocketMQTemplate")
public class RocketMQTemplateController {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Value("${rocketmq.pull-consumer.namespace}")
    private String pullConsumerNamespace;

    @GetMapping("convertAndSend")
    public void convertAndSend() {
        //send message synchronously
        rocketMQTemplate.convertAndSend("convertAndSend-string", "Hello, World!");
        //send spring message
        Message<String> message = MessageBuilder.withPayload("Hello, World! I'm from spring message").build();
        rocketMQTemplate.send("send-message", message);
        //send messgae asynchronously
        rocketMQTemplate.asyncSend("asyncSend-object", new OrderPaidEvent("T_001", new BigDecimal("88.00")), new SendCallback() {
            @Override
            public void onSuccess(SendResult var1) {
                System.out.printf("async onSucess SendResult=%s %n", var1);
            }

            @Override
            public void onException(Throwable var1) {
                System.out.printf("async onException Throwable=%s %n", var1);
            }

        });
        //Send messages orderly
        rocketMQTemplate.syncSendOrderly("syncSendOrderly-message", MessageBuilder.withPayload("Hello, World").build(), "hashkey");

    }

    @GetMapping("sendAndReceive")
    public void sendAndReceive() {
        // 同步发送request并且等待String类型的返回值
        // 请求超时时间 默认值：3000毫秒。:send request message to <sendAndReceive-string-string> OK, but wait reply message timeout, 3000 ms.
        String replyString = rocketMQTemplate.sendAndReceive("sendAndReceive-string-string", "request string", String.class);
        System.out.printf("send %s and receive %s %n", "request string", replyString);
        // 异步发送request并且等待User类型的返回值
        //要求在3000ms内进入onSuccess：RequestTimeoutException: CODE: 10006  DESC: request timeout, no reply message.
        rocketMQTemplate.sendAndReceive("sendAndReceive-user-user", new User("requestUserName", (byte) 9), new RocketMQLocalRequestCallback<User>() {
            @Override
            public void onSuccess(User message) {
                System.out.printf("send user object and receive %s %n", message.toString());
            }

            @Override
            public void onException(Throwable e) {
                //onSuccess要求在5000ms结束：onException - Consumer replies message failed. error: invokeAsync call the addr[192.168.129.97:10911] timeout
                e.printStackTrace();
            }
        }, 5000);
    }

    /**
     * 发送广播消息
     */
    @RequestMapping("/testBroadSend")
    public void testSyncSend() {
        for (int i = 0; i < 2; i++) {
            rocketMQTemplate.convertAndSend("broadcasting-string", "test-message" + i);
        }
        for (int i = 0; i < 2; i++) {
            rocketMQTemplate.convertAndSend("clustering-string", "test-message" + i);
        }
    }

    @GetMapping("testPull")
    public void testPull() throws ClientException, IOException {
        //send message synchronously
        rocketMQTemplate.convertAndSend("pull", "pull: Hello, World!");
    }

    @GetMapping("testTrace")
    public void testTrace() throws ClientException, IOException {
        //send message synchronously
        rocketMQTemplate.convertAndSend("convertAndSend-string-trace", "pull: Hello, World!");
    }

    @GetMapping("testTag")
    public void testTag() throws ClientException, IOException {
        //send message synchronously
        rocketMQTemplate.convertAndSend("convertAndSend-string-tage:pay", "pull: Hello, World!");
    }

    @GetMapping("testHeader")
    public void testHeader() throws ClientException, IOException {
        //send message synchronously
        Map<String, Object> headers = new HashMap<>();
        headers.put("name", "alex");
        rocketMQTemplate.convertAndSend("convertAndSend-string-header", "pull: Hello, World!",headers);
    }


    @GetMapping("receive")
    public void receive() {
        //从单一主题和消费者组拉取，默认拉取一条
        List<String> messages = rocketMQTemplate.receive(String.class);
        System.out.printf("receive from rocketMQTemplate, messages=%s %n", messages);
    }


    @GetMapping("/sendTransaction")
    public void sendTransaction() {
        String destination = "transaction_topic:tx";
        for (int i = 1; i < 6; i++) {
            Message<String> message = MessageBuilder.withPayload(String.format("事务消息%s", i))
                    .setHeader("topic", "transaction_topic:tx")
                    .setHeader("orderId", i)
                    .build();
            // 发送事务消息
            TransactionSendResult res = rocketMQTemplate.sendMessageInTransaction(destination, message, null);
            log.debug("msgId = {} , sendStatus = {}", res.getMsgId(), res.getSendStatus());
        }

    }





}
